package com.test;

import com.alibaba.fastjson.JSON;
import com.test.entry.Order;
import com.test.source.RocketMQConsumer;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

public class EventTimeJob {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置时间的特征
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStreamSource<String> streamSource = env.addSource(new RocketMQConsumer("FLINK-TEST-TOPIC"));

        SingleOutputStreamOperator<Order> dataStream = streamSource
                .filter(s -> StringUtils.isNotEmpty(s))
                .map(s -> JSON.parseObject(s, Order.class))
                .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<Order>() {
                    @Override
                    public long extractAscendingTimestamp(Order order) {
                        return order.getCreateDate().getTime();
                    }
                }).
                        name("流转商品 pojo");

        SingleOutputStreamOperator<Tuple2<String, Integer>> operator = dataStream.flatMap(new Tokenizer());

        operator
                .keyBy(0)
                // 如果是时间的特征是事件时间，那么此时窗口所使用的时间是事件时间，而不是系统时间
                .timeWindow(Time.minutes(1))
                //.allowedLateness(Time.seconds(10))
                .sum(1)
                /* .reduce(new ReduceFunction<Order>() {
                     @Override
                     public Order reduce(Order order, Order t1) throws Exception {
                         Order o = new Order();
                         o.setAccountNo(order.getAccountNo());
                         o.setAmount(order.getAmount().add(t1.getAmount()));
                         o.setName(order.getName());
                         return o;
                     }
                 })*/.print();

        env.execute("事件时间示例");

    }

    // User-defined functions
    public static class Tokenizer implements FlatMapFunction<Order, Tuple2<String, Integer>> {

        @Override
        public void flatMap(Order value, Collector<Tuple2<String, Integer>> out) {
            // emit the pairs
            System.out.println("key:" + value.getAccountNo());
            out.collect(new Tuple2(value.getAccountNo(), 1));
        }
    }
}
